
<!DOCTYPE html>
<html lang="zh-CN">

<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1, maximum-scale=1, user-scalable=no">
  <meta http-equiv="X-UA-Compatible" content="ie=edge">
  <title>RabbitMQ消息中间件 - 我在开了灯的床头下，想问问自己的心啊。</title>
  <meta name="apple-mobile-web-app-capable" content="yes" />
  <meta name="apple-mobile-web-app-status-bar-style" content="black-translucent">
  <meta name="google" content="notranslate" />
  
  <meta name="keywords" content="yuanyl,"> 
  
  <meta name="description" content="The Technology blog,RabbitMQ消息中间件一、消息中间件介绍
我们用j..."> 
  
  <meta name="author" content="yuanyl"> 
  <link rel="shortcut icon" href="http://freeyjay.gitee.io/blog/image/favicon.ico">
  
<link rel="stylesheet" href="/blog/css/style.css">


  
  
<link rel="stylesheet" href="//at.alicdn.com/t/font_1445822_58xq2j9v1id.css">

  

  
  
  
<link rel="stylesheet" href="https://cdn.bootcss.com/fancybox/3.5.7/jquery.fancybox.min.css">

  

  
  
  
<link rel="stylesheet" href="https://cdn.bootcss.com/highlight.js/9.15.10/styles/a11y-dark.min.css">

  

  <script>
    var CONFIG = window.CONFIG || {}
    CONFIG = {
      fancybox: true,
      pjax: false,
      donate_alipay: '',
      donate_wechat: 'http://freeyjay.gitee.io/image_bed/pic/wechat.jpg',
      gitalk: {
        enable: false,
        clientID: '',
        clientSecret: '',
        id: window.location.pathname,
        repo: '',
        owner: '',
        admin: ''
      },
      motto: {
        api: '',
        default: '橘子金黄，玻璃晴朗。'
      }
    }
  </script>

  

<meta name="generator" content="Hexo 4.2.0"></head>
<body class="lock-screen">
  <div class="loading">
  <div class="loading-wrap">
    <div class="loading-spinner">
      <div class="loading-bounce1"></div>
      <div class="loading-bounce2"></div>
    </div>
  </div>
</div>
  <header>
  <nav class="menu">
  <div class="menu-close">
    <i class="iconfont iconplus"></i>
  </div>
  <ul class="menu-content">
    
    
    
    
    <li class="menu-item"><a href="/ "> 首页</a></li>
    
    
    
    
    <li class="menu-item"><a href="/archives "> 归档</a></li>
    
    
    
    
    <li class="menu-item"><a href="/about "> 关于</a></li>
    
  </ul>
  <div class="menu-copyright"><p>Copyright© 2019-2019 | <a target="_blank" href="https://gitee.com/FreeYJay">yuanyl</a> .AllRightsReserved</p></div>
</nav>
</header>
  <main id="main">
  <div class="container" id="container">
    <article class="article">
  <section class="head">
  <img src="http://freeyjay.gitee.io/blog/image/post-image.jpg" draggable="false">
  <div class="head-mask">
    <h1 class="head-title">RabbitMQ消息中间件</h1>
    <div class="head-info">
      <span class="post-info-item"><i class="iconfont iconcalendar"></i>2020-04-30</span
        class="post-info-item">
      
      <span class="post-info-item"><i class="iconfont iconfont-size"></i>8505</span>
    </div>
  </div>
</section>
  <section class="main">
    <section class="content">
      <h1 id="RabbitMQ消息中间件"><a href="#RabbitMQ消息中间件" class="headerlink" title="RabbitMQ消息中间件"></a>RabbitMQ消息中间件</h1><h2 id="一、消息中间件介绍"><a href="#一、消息中间件介绍" class="headerlink" title="一、消息中间件介绍"></a>一、消息中间件介绍</h2><ol>
<li><p>我们用java来举例子， 打个比方 我们客户端发送一个下单请求给订单系统（order）订单系统发送了 一个请求给我们的库存系统告诉他需要更改库存了， 我已经下单了， 这里， 每一个请求我们都可以看作一条消息，但是 我们客户端需要等待订单系统告诉我这条消息的处理结果（我到底有没有下单成功） 但是 订单系统不需要知道库存系统这条消息的处理情况 因为无论你库存有没有改动成功， 我订单还是下了， 因为是先下完了订单（下成功了） 才去更改库存， 库存如果更改出BUG了 那是库存系统的问题， 这个BUG不会影响订单系统。如果这里你能理解的话， 那么我们就能发现 我们用户发送的这条消息（下订单），是需要同步的（我需要知道结果）， 订单发送给库存的消息，是可以异步的（我不想知道你库存到底改了没，我只是通知你我这边成功下了一个订单）。</p>
<p>那么如果我们还按原来的方式去实现这个需求的话， 那么结果会是这样：</p>
<p><img src="http://freeyjay.gitee.io/image_bed/pic/MQ1.png" alt="image-20200430132621164"></p>
<p>在实际使用的时候，我们也可以使用线程池处理我们的订单入库存的步骤，但是线程池会有线程池阻塞等缺点。</p>
<p>我们可以采用如下设计方式实现异步存储入库操作。</p>
<p><img src="http://freeyjay.gitee.io/image_bed/pic/MQ2.png" alt="image-20200430132722108"></p>
</li>
</ol>
<h2 id="二、RabbitMQ介绍以及AMQP协议"><a href="#二、RabbitMQ介绍以及AMQP协议" class="headerlink" title="二、RabbitMQ介绍以及AMQP协议"></a>二、RabbitMQ介绍以及AMQP协议</h2><ol>
<li><p>AMQP 协议中的基本概念：</p>
<p>•     <strong>Broker</strong>: 接收和分发消息的应用，我们在介绍消息中间件的时候所说的消息系统就是Message Broker。</p>
<p>•     <strong>Virtual host</strong>: 出于多租户和安全因素设计的，把AMQP的基本组件划分到一个虚拟的分组中，类似于网络中的namespace概念。当多个不同的用户使用同一个RabbitMQ server提供的服务时，可以划分出多个vhost，每个用户在自己的vhost创建exchange／queue等。</p>
<p>•     <strong>Connection</strong>: publisher／consumer和broker之间的TCP连接。断开连接的操作只会在client端进行，Broker不会断开连接，除非出现网络故障或broker服务出现问题。</p>
<p>•     <strong>Channel</strong>: 如果每一次访问RabbitMQ都建立一个Connection，在消息量大的时候建立TCP Connection的开销将是巨大的，效率也较低。Channel是在connection内部建立的逻辑连接，如果应用程序支持多线程，通常每个thread创建单独的channel进行通讯，AMQP method包含了channel id帮助客户端和message broker识别channel，所以channel之间是完全隔离的。Channel作为轻量级的Connection极大减少了操作系统建立TCP connection的开销。</p>
<p>•     <strong>Exchange</strong>: message到达broker的第一站，根据分发规则，匹配查询表中的routing key，分发消息到queue中去。常用的类型有：direct (point-to-point), topic (publish-subscribe) and fanout (multicast)。</p>
<p>•     <strong>Queue</strong>: 消息最终被送到这里等待consumer取走。一个message可以被同时拷贝到多个queue中。</p>
<p>•     <strong>Binding</strong>: exchange和queue之间的虚拟连接，binding中可以包含routing key。Binding信息被保存到exchange中的查询表中，用于message的分发依据。</p>
<h3 id="Exchange的类型"><a href="#Exchange的类型" class="headerlink" title="Exchange的类型"></a>Exchange的类型</h3><h4 id="direct"><a href="#direct" class="headerlink" title="direct :"></a>direct :</h4><p>​       这种类型的交换机的路由规则是根据一个routingKey的标识，交换机通过一个routingKey与队列绑定 ，在生产者生产消息的时候 指定一个routingKey 当绑定的队列的routingKey 与生产者发送的一样 那么交换机会吧这个消息发送给对应的队列。</p>
<h4 id="fanout"><a href="#fanout" class="headerlink" title="fanout:"></a>fanout:</h4><p>​       这种类型的交换机路由规则很简单，只要与他绑定了的队列， 他就会吧消息发送给对应队列（与routingKey没关系）</p>
<h4 id="topic"><a href="#topic" class="headerlink" title="topic:"></a>topic:</h4><p>​       这种类型的交换机路由规则也是和routingKey有关 只不过 topic他可以根据:星,#（ 星号代表过滤一单词，#代表过滤后面所有单词， 用.隔开）来识别routingKey 我打个比方 假设 我绑定的routingKey 有队列A和B A的routingKey是：星.user B的routingKey是: #.user</p>
<p>​      那么我生产一条消息routingKey 为： error.user 那么此时 2个队列都能接受到， 如果改为 topic.error.user 那么这时候 只有B能接受到了</p>
<h4 id="headers"><a href="#headers" class="headerlink" title="headers:"></a>headers:</h4><p>​       这个类型的交换机很少用到，他的路由规则 与routingKey无关 而是通过判断header参数来识别的， 基本上没有应用场景，因为上面的三种类型已经能应付了。</p>
</li>
<li><p>主流MQ对比</p>
<p><img src="http://freeyjay.gitee.io/image_bed/pic/MQ3.png" alt="image-20200430133349866"></p>
</li>
</ol>
<h2 id="三、Springboot整合RabbitMQ"><a href="#三、Springboot整合RabbitMQ" class="headerlink" title="三、Springboot整合RabbitMQ"></a>三、Springboot整合RabbitMQ</h2><p>引入依赖</p>
<figure class="highlight"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br></pre></td><td class="code"><pre><span class="line">&lt;dependency&gt;</span><br><span class="line">    &lt;groupId&gt;org.springframework.boot&lt;/groupId&gt;</span><br><span class="line">    &lt;artifactId&gt;spring-boot-starter-amqp&lt;/artifactId&gt;</span><br><span class="line">&lt;/dependency&gt;</span><br></pre></td></tr></table></figure>

<p>配置连接、交换机、队列</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br></pre></td><td class="code"><pre><span class="line"> <span class="meta">@Bean</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> ConnectionFactory <span class="title">connectionFactory</span><span class="params">()</span> </span>&#123;</span><br><span class="line">    CachingConnectionFactory connectionFactory = <span class="keyword">new</span></span><br><span class="line">CachingConnectionFactory(<span class="string">"localhost"</span>,<span class="number">5672</span>); <span class="comment">//我这里直接在构造方法传入了</span></span><br><span class="line"><span class="comment">// connectionFactory.setHost();</span></span><br><span class="line"><span class="comment">// connectionFactory.setPort(); connectionFactory.setUsername("guest"); connectionFactory.setPassword("guest"); connectionFactory.setVirtualHost("testhost"); </span></span><br><span class="line">  <span class="comment">//是否开启消息确认机制</span></span><br><span class="line">  <span class="comment">//connectionFactory.setPublisherConfirms(true);</span></span><br><span class="line">  <span class="keyword">return</span> connectionFactory;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p>创建交换机并绑定队列</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br></pre></td><td class="code"><pre><span class="line"> <span class="meta">@Bean</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> DirectExchange  <span class="title">defaultExchange</span><span class="params">()</span> </span>&#123;</span><br><span class="line">    <span class="keyword">return</span> <span class="keyword">new</span> DirectExchange(<span class="string">"directExchange"</span>);</span><br><span class="line">&#125;</span><br><span class="line"><span class="meta">@Bean</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> Queue <span class="title">queue</span><span class="params">()</span> </span>&#123; <span class="comment">//名字 是否持久化</span></span><br><span class="line">    <span class="keyword">return</span> <span class="keyword">new</span> Queue(<span class="string">"testQueue"</span>, <span class="keyword">true</span>);</span><br><span class="line">&#125;</span><br><span class="line"><span class="meta">@Bean</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> Binding <span class="title">binding</span><span class="params">()</span> </span>&#123;</span><br><span class="line"><span class="comment">//绑定一个队列 to: 绑定到哪个交换机上面 with:绑定的路由建(routingKey)</span></span><br><span class="line"><span class="keyword">return</span> BindingBuilder.bind(queue()).to(defaultExchange()).with(<span class="string">"direct.key"</span>);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p>发送消息模板</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Bean</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> RabbitTemplate <span class="title">rabbitTemplate</span><span class="params">(ConnectionFactory connectionFactory)</span> </span>&#123;</span><br><span class="line"><span class="comment">//注意 这个ConnectionFactory 是使用javaconfig方式配置连接的时候才需要传入的 如果是yml配置的连接的话是不需要的</span></span><br><span class="line">RabbitTemplate template = <span class="keyword">new</span> RabbitTemplate(connectionFactory);</span><br><span class="line">    <span class="keyword">return</span> template;</span><br><span class="line">&#125;</span><br><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">TestSend</span> </span>&#123;</span><br><span class="line">    <span class="meta">@Autowired</span></span><br><span class="line">    RabbitTemplate rabbitTemplate;</span><br><span class="line"><span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">testSend</span><span class="params">()</span> </span>&#123;</span><br><span class="line"><span class="comment">//至于为什么调用这个API 后面会解释</span></span><br><span class="line"><span class="comment">//参数介绍: 交换机名字，路由建， 消息内容 		  </span></span><br><span class="line">  rabbitTemplate.convertAndSend(<span class="string">"directExchange"</span>, <span class="string">"direct.key"</span>, <span class="string">"hello"</span>);</span><br><span class="line">&#125; </span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p>接收消息</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">TestListener</span>  </span>&#123;</span><br><span class="line"> <span class="meta">@RabbitListener</span>(queues = <span class="string">"testQueue"</span>)</span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">get</span><span class="params">(String message)</span> <span class="keyword">throws</span> Exception</span>&#123;</span><br><span class="line">        System.out.println(message);</span><br><span class="line">    &#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<h2 id="四、可靠性保证"><a href="#四、可靠性保证" class="headerlink" title="四、可靠性保证"></a>四、可靠性保证</h2><ol>
<li><p>生产者</p>
<p>不确定消息有没有发送到Broker，可以采用失败回调，发送方确认模式</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Bean</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> RabbitTemplate <span class="title">rabbitTemplate</span><span class="params">(ConnectionFactory connectionFactory)</span> </span>&#123; RabbitTemplate template = <span class="keyword">new</span> RabbitTemplate(connectionFactory); <span class="comment">//开启mandatory模式(开启失败回调)</span></span><br><span class="line">template.setMandatory(<span class="keyword">true</span>);</span><br><span class="line"><span class="comment">//指定失败回调接口的实现类 </span></span><br><span class="line">                                                                           template.setReturnCallback(<span class="keyword">new</span> MyReturnCallback()); <span class="keyword">return</span> template;</span><br><span class="line">&#125;</span><br><span class="line"></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">MyReturnCallback</span> <span class="keyword">implements</span> <span class="title">RabbitTemplate</span>.<span class="title">ReturnCallback</span> </span>&#123;</span><br><span class="line"><span class="meta">@Override</span></span><br><span class="line"> <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">returnedMessage</span><span class="params">(Message message, <span class="keyword">int</span> replyCode, String replyText,String exchange, String routingKey)</span> </span>&#123;</span><br><span class="line">&#125; &#125;</span><br><span class="line">System.out.println(message);</span><br><span class="line">System.out.println(replyCode);</span><br><span class="line">System.out.println(replyText);</span><br><span class="line">System.out.println(exchange);</span><br><span class="line">System.out.println(routingKey);</span><br><span class="line">&#125;</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p>当指定的交换机不能吧消息路由到队列时(没有指定路由建或者指定的RouteKey没有绑定对应的队列或者压根就没有绑定队列都会失败) 消息就会发送失败</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">//开启发送确认模式	</span></span><br><span class="line">connectionFactory.setPublisherConfirms(<span class="keyword">true</span>);</span><br><span class="line"></span><br><span class="line"></span><br><span class="line"><span class="comment">//在rabbitmqTemplate配置一下</span></span><br><span class="line">template.setConfirmCallback(<span class="keyword">new</span> MyConfirmCallback());</span><br><span class="line"></span><br><span class="line"><span class="comment">//失败回调</span></span><br><span class="line"> <span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">MyConfirmCallback</span> <span class="keyword">implements</span> <span class="title">RabbitTemplate</span>.<span class="title">ConfirmCallback</span></span>&#123;</span><br><span class="line"><span class="meta">@Override</span></span><br><span class="line">   <span class="comment">//业务id，是否发送成功，原因</span></span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">confirm</span><span class="params">(CorrelationData correlationData, <span class="keyword">boolean</span> ack, String cause)</span> </span>&#123;</span><br><span class="line">        System.out.println(correlationData);</span><br><span class="line">        System.out.println(ack);</span><br><span class="line">       System.out.println(cause);</span><br><span class="line">&#125; &#125;</span><br></pre></td></tr></table></figure>

<p>为了进一步保证我们交换机的可靠性，我们可以定义一个备用交换机进行绑定主交换机，当主交换机不可用时，会自动进入备用交换机，一般情况备用交换机都是采用fanout类型。</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Bean</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> DirectExchange <span class="title">defaultExchange</span><span class="params">()</span> </span>&#123;</span><br><span class="line">Map&lt;String, Object&gt; map = <span class="keyword">new</span> HashMap&lt;&gt;();</span><br><span class="line"> <span class="comment">//备用交换机名称</span></span><br><span class="line"> map.put(<span class="string">"alternate-exchange"</span>, <span class="string">"name"</span>);</span><br><span class="line"> <span class="keyword">return</span> <span class="keyword">new</span> DirectExchange(<span class="string">"directExchangeTest4"</span>, <span class="keyword">false</span>, <span class="keyword">false</span>, map);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>



</li>
</ol>
<ol start="2">
<li><p>消费者</p>
<h3 id="手动确认"><a href="#手动确认" class="headerlink" title="手动确认"></a>手动确认</h3><p>为什么要确认消费? 默认情况下消费者在拿到rabbitmq的消息时 已经自动确认这条消息已经消费了， 讲白话就 是rabbitmq的队列里就会删除这条消息了， 但是 我们实际开发中 难免会遇到这种情况， 比如说 拿到这条消息发现我处理不了比如说参数不对， 又比如说 我当前这个系统出问题了， 暂时不能处理这个消息， 但是这个消息已 经被你消费掉了 rabbitmq的队列里也删除掉了， 你自己这边又处理不了， 那么这个消息就被遗弃了。 这种情 况在实际开发中是不合理的， rabbitmq提供了解决这个问题的方案， 也就是我们上面所说的confirm模式 只是我 们刚刚讲的是发送方的 这次我们来讲消费方的。 </p>
<p>​    </p>
<p>配置手动确认：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br></pre></td><td class="code"><pre><span class="line"> <span class="meta">@Bean</span></span><br><span class="line"><span class="keyword">public</span> SimpleRabbitListenerContainerFactory</span><br><span class="line">simpleRabbitListenerContainerFactory(ConnectionFactory connectionFactory)&#123;</span><br><span class="line">    SimpleRabbitListenerContainerFactory simpleRabbitListenerContainerFactory =</span><br><span class="line">            <span class="keyword">new</span> SimpleRabbitListenerContainerFactory();</span><br><span class="line"><span class="comment">//这个connectionFactory就是我们自己配置的连接工厂直接注入进来 simpleRabbitListenerContainerFactory.setConnectionFactory(connectionFactory); </span></span><br><span class="line">  <span class="comment">//这边设置消息确认方式由自动确认变为手动确认 simp</span></span><br></pre></td></tr></table></figure>

<p>消费者手动确认：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br><span class="line">12</span><br><span class="line">13</span><br><span class="line">14</span><br><span class="line">15</span><br><span class="line">16</span><br><span class="line">17</span><br><span class="line">18</span><br><span class="line">19</span><br><span class="line">20</span><br><span class="line">21</span><br></pre></td><td class="code"><pre><span class="line"><span class="meta">@Component</span></span><br><span class="line"><span class="keyword">public</span> <span class="class"><span class="keyword">class</span> <span class="title">TestListener</span>  </span>&#123;</span><br><span class="line"><span class="comment">//containerFactory:指定我们刚刚配置的容器 @RabbitListener(queues = "testQueue",containerFactory =</span></span><br><span class="line"><span class="string">"simpleRabbitListenerContainerFactory"</span>)</span><br><span class="line">    <span class="function"><span class="keyword">public</span> <span class="keyword">void</span> <span class="title">getMessage</span><span class="params">(Message message, Channel channel)</span> <span class="keyword">throws</span> Exception</span>&#123;</span><br><span class="line">System.out.println(<span class="keyword">new</span> String(message.getBody(),<span class="string">"UTF-8"</span>)); </span><br><span class="line">  System.out.println(message.getBody());</span><br><span class="line"><span class="comment">//这里我们调用了一个下单方法 如果下单成功了 那么这条消息就可以确认被消费了，即实际业务处理类 </span></span><br><span class="line">  <span class="keyword">boolean</span> f =placeAnOrder();</span><br><span class="line"><span class="keyword">if</span> (f)&#123;</span><br><span class="line"><span class="comment">//传入这条消息的标识， 这个标识由rabbitmq来维护 我们只需要从message中拿出来就可以 </span></span><br><span class="line">  <span class="comment">//第二个boolean参数指定是不是批量处理的 什么是批量处理我们待会儿会讲到 channel.basicAck(message.getMessageProperties().getDeliveryTag(),false);</span></span><br><span class="line">&#125;<span class="keyword">else</span> &#123;</span><br><span class="line"><span class="comment">//当然 如果这个订单处理失败了 我们也需要告诉rabbitmq 告诉他这条消息处理失败了 可以退回</span></span><br><span class="line">也可以遗弃 要注意的是 无论这条消息成功与否 一定要通知 就算失败了 如果不通知的话 rabbitmq端会显示这条 消息一直处于未确认状态，那么这条消息就会一直堆积在rabbitmq端 除非与rabbitmq断开连接 那么他就会把这条 消息重新发给别人 所以 一定要记得通知!</span><br><span class="line"><span class="comment">//前两个参数 和上面的意义一样， 最后一个参数 就是这条消息是返回到原队列 还是这条消息作废 就是不退回了。</span></span><br><span class="line">channel.basicNack(message.getMessageProperties().getDeliveryTag(),<span class="keyword">false</span>,<span class="keyword">true</span>);</span><br><span class="line"><span class="comment">//其实 这个API也可以去告诉rabbitmq这条消息失败了 与basicNack不同之处 就是 他不能批量</span></span><br><span class="line">处理消息结果 只能处理单条消息 其实basicNack作为basicReject的扩展开发出来的</span><br><span class="line"> <span class="comment">//channel.basicReject(message.getMessageProperties().getDeliveryTag(),true);</span></span><br><span class="line">&#125; &#125;</span><br></pre></td></tr></table></figure>

<h3 id="消息预取"><a href="#消息预取" class="headerlink" title="消息预取"></a>消息预取</h3><p>rabbitmq 默认 他会最快 以轮询的机制吧队列所有的消息发送给所有客户端 </p>
<p>那么 这种机制会有什么问题呢， 对于Rabbitmq来讲 这样子能最快速的使自己不会囤积消息而对性能造成影响， 但是 对于我们整个系统来讲， 这种机制会带来很多问题， 比如说 我一个队列有2个人同时在消费，而且他们处理 能力不同， 我打个最简单的比方 有100个订单消息需要处理(消费) 现在有消费者A 和消费者B ， 消费者A消费一 条消息的速度是 10ms 消费者B 消费一条消息的速度是15ms ( 当然 这里只是打比方) 那么 rabbitmq 会默认给 消费者A B 一人50条消息让他们消费 但是 消费者A 他500ms 就可以消费完所有的消息 并且处于空闲状态 而 消费 者B需要750ms 才能消费完 如果从性能上来考虑的话 这100条消息消费完的时间一共是750ms(因为2个人同时在 消费) 但是如果 在消费者A消费完的时候 能把这个空闲的性能用来和B一起消费剩下的信息的话， 那么这处理速 度就会快非常多。 </p>
<p>配置消息预取：</p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br></pre></td><td class="code"><pre><span class="line"><span class="comment">//设置消息预取的数量1-2500，500平均值</span></span><br><span class="line">simpleRabbitListenerContainerFactory.setPrefetchCount(<span class="number">1</span>);</span><br></pre></td></tr></table></figure>

<h3 id="死信交换机"><a href="#死信交换机" class="headerlink" title="死信交换机"></a>死信交换机</h3><p>在创建队列的时候 可以给这个队列附带一个交换机， 那么这个队列作废的消息就会被重 新发到附带的交换机，然后让这个交换机重新路由这条消息 </p>
<figure class="highlight java"><table><tr><td class="gutter"><pre><span class="line">1</span><br><span class="line">2</span><br><span class="line">3</span><br><span class="line">4</span><br><span class="line">5</span><br><span class="line">6</span><br><span class="line">7</span><br><span class="line">8</span><br><span class="line">9</span><br><span class="line">10</span><br><span class="line">11</span><br></pre></td><td class="code"><pre><span class="line"> <span class="meta">@Bean</span></span><br><span class="line"><span class="function"><span class="keyword">public</span> Queue <span class="title">queue</span><span class="params">()</span> </span>&#123;</span><br><span class="line">Map&lt;String,Object&gt; map = <span class="keyword">new</span> HashMap&lt;&gt;();</span><br><span class="line"><span class="comment">//设置消息的过期时间 单位毫秒</span></span><br><span class="line">map.put(<span class="string">"x-message-ttl"</span>,<span class="number">10000</span>);</span><br><span class="line"><span class="comment">//设置附带的死信交换机</span></span><br><span class="line">map.put(<span class="string">"x-dead-letter-exchange"</span>,<span class="string">"exchange.dlx"</span>);</span><br><span class="line"><span class="comment">//指定重定向的路由建 消息作废之后可以决定需不需要更改他的路由建 如果需要 就在这里指定 </span></span><br><span class="line"> map.put(<span class="string">"x-dead-letter-routing-key"</span>,<span class="string">"dead.order"</span>);</span><br><span class="line"><span class="keyword">return</span> <span class="keyword">new</span> Queue(<span class="string">"testQueue"</span>, <span class="keyword">true</span>,<span class="keyword">false</span>,<span class="keyword">false</span>,map);</span><br><span class="line">&#125;</span><br></pre></td></tr></table></figure>

<p>大致效果如图</p>
<p><img src="http://freeyjay.gitee.io/image_bed/pic/MQ4.png" alt="image-20200430144227781"></p>
</li>
</ol>

    </section>
    <section class="extra">
      
        <section class="donate">
  <div class="qrcode">
    <img src="">
  </div>
  <div class="icon">
    <a href="javascript:;" id="alipay"><i class="iconfont iconalipay"></i></a>
    <a href="javascript:;" id="wechat"><i class="iconfont iconwechat-fill"></i></a>
  </div>
</section>
      
      
      
<nav class="nav">
  
  
    <a href="/blog/2020/04/27/K8S%E6%90%AD%E5%BB%BA%E8%B8%A9%E5%9D%91%E4%B9%8B%E8%B7%AF/">K8S搭建踩坑之路<i class="iconfont iconright"></i></a>
  
</nav>

    </section>
    
  </section>
</article>
  </div>
</main>
  <footer class="footer">
  <div class="footer-social">
    
    
    
    
    
    <a href="tencent://message/?Menu=yes&uin=1095159827 " target="_blank" onMouseOver="this.style.color= '#12B7F5'"
      onMouseOut="this.style.color='#33333D'">
      <i class="iconfont footer-social-item  iconQQ "></i>
    </a>
    
    
    
    
    
    <a href="javascript:; " target="_blank" onMouseOver="this.style.color= '#09BB07'"
      onMouseOut="this.style.color='#33333D'">
      <i class="iconfont footer-social-item  iconwechat-fill "></i>
    </a>
    
    
    
    
    
    <a href="https://www.instagram.com/yuangege233/ " target="_blank" onMouseOver="this.style.color= '#DA2E76'"
      onMouseOut="this.style.color='#33333D'">
      <i class="iconfont footer-social-item  iconinstagram "></i>
    </a>
    
    
    
    
    
    <a href="https://github.com/yuanyonglin " target="_blank" onMouseOver="this.style.color= '#24292E'"
      onMouseOut="this.style.color='#33333D'">
      <i class="iconfont footer-social-item  icongithub-fill "></i>
    </a>
    
    
    
    
    
    <a href="mailto:yuanyonglin1995@gmail.com " target="_blank" onMouseOver="this.style.color='#FFBE5B'"
      onMouseOut="this.style.color='#33333D'">
      <i class="iconfont footer-social-item  iconmail"></i>
    </a>
    
  </div>
  <div class="footer-copyright"><p>Copyright© 2019-2019 | <a target="_blank" href="https://gitee.com/FreeYJay">yuanyl</a> .AllRightsReserved</p></div>
</footer>
    <div class="fab fab-plus">
    <i class="iconfont iconplus"></i>
  </div>
  <div class="fab fab-up">
    <i class="iconfont iconcaret-up"></i>
  </div>
  <div class="fab fab-menu">
    <i class="iconfont iconmenu"></i>
  </div>
</body>


<script src="https://cdn.bootcss.com/jquery/3.4.1/jquery.min.js"></script>






<script src="https://cdn.bootcss.com/fancybox/3.5.7/jquery.fancybox.min.js"></script>








<script src="/blog/js/script.js"></script>



</html>